package com.neusoft.bizcore.notice.notification.kafka;

import java.util.Date;
import java.util.concurrent.ExecutorService;

import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.neusoft.bizcore.mom.bean.MessageBean;
import com.neusoft.bizcore.notice.notification.dto.NoticeRecordDTO;
import com.neusoft.bizcore.notice.notification.dto.NoticeRecordMailDTO;
import com.neusoft.bizcore.notice.notification.dto.NoticeRecordSystemDTO;
import com.neusoft.bizcore.notice.notification.service.NoticeService;

import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
@ConditionalOnProperty(value = "spring.kafka.bootstrap-servers")
public class NoticeConsumer {
    @Autowired
    private NoticeService noticeService;

    @Resource(name = "noticeKafkaExecutor")
    ExecutorService noticeKafkaExecutor;

    @KafkaListener(topics = {
            "${notice.kafka.receive-topic-name:notice}" }, group = "${spring.kafka.consumer.group-id:notice-consumer}")
    public void listen(final MessageBean messageBean) throws Exception {
        NoticeConsumer.log.info("kafka收到的消息" + messageBean.toString());
        final SendMessageRunnable runnable = new SendMessageRunnable(messageBean);
        this.noticeKafkaExecutor.execute(runnable);
    }

    private NoticeRecordDTO getNoticeRecordDTO(final JSONObject obj, final Date sendDate) {
        final NoticeRecordDTO dto = new NoticeRecordDTO();
        dto.setAppCode(obj.getString("appCode"));
        dto.setCreateTime(sendDate == null ? new Date() : sendDate);
        dto.setType(obj.getJSONArray("type").toJavaList(String.class));
        dto.setTitle(obj.getString("title"));
        dto.setMail(this.getNoticeRecordMailDTO(obj.getJSONObject("mail")));
        dto.setSystem(this.getNoticeRecordSystemDTO(obj.getJSONObject("system")));
        return dto;
    }

    private NoticeRecordMailDTO getNoticeRecordMailDTO(final JSONObject obj) {

        if (obj != null) {
            final NoticeRecordMailDTO dto = new NoticeRecordMailDTO();
            dto.setEmailAddress(obj.getString("emailAddress"));
            dto.setTemplateMsg(obj.getString("templateMsg"));
            dto.setTemplateName(obj.getString("templateName"));
            return dto;
        }
        return null;

    }

    private NoticeRecordSystemDTO getNoticeRecordSystemDTO(final JSONObject obj) {
        if (obj != null) {
            final NoticeRecordSystemDTO dto = new NoticeRecordSystemDTO();
            dto.setBusinessInfo(obj.getString("businessInfo"));
            dto.setContent(obj.getString("content"));
            dto.setNoticeType(obj.getString("noticeType"));
            dto.setNoticeRoute(obj.getString("noticeRoute"));
            final JSONArray array = obj.getJSONArray("users");
            if (!array.isEmpty()) {
                dto.setUsers(array.toJavaList(String.class));
            }
            return dto;
        }
        return null;
    }

    class SendMessageRunnable implements Runnable {

        MessageBean messageBean;

        public SendMessageRunnable(final MessageBean messageBean) {
            this.messageBean = messageBean;
        }

        @Override
        public void run() {
            NoticeConsumer.this.noticeService
                    .buildAndSendMessage(
                            NoticeConsumer.this.getNoticeRecordDTO(JSON.parseObject(this.messageBean.getMessage()),
                                    this.messageBean.getSendDate()));

        }

    }

}
